package com.neusoft.databus.policy.service.impl;

import java.util.Date;
import java.util.List;

import javax.annotation.PostConstruct;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.collect.Lists;
import com.google.common.eventbus.AsyncEventBus;
import com.google.common.eventbus.Subscribe;
import com.neusoft.bizcore.common.bean.MessageBean;
import com.neusoft.bizcore.common.cache.CacheService;
import com.neusoft.bizcore.common.messaging.ProducerService;
import com.neusoft.bizcore.web.utils.JsonUtils;
import com.neusoft.databus.policy.bean.ComplexPolicyExtendBean;
import com.neusoft.databus.policy.constants.DatabusPolicyConstant;
import com.neusoft.databus.policy.converter.EventConverter;
import com.neusoft.databus.policy.enums.EventStatus;
import com.neusoft.databus.policy.enums.PolicyType;
import com.neusoft.databus.policy.model.Event;
import com.neusoft.databus.policy.model.EventAttribute;
import com.neusoft.databus.policy.model.Policy;
import com.neusoft.databus.policy.repository.EventAttibuteRepository;
import com.neusoft.databus.policy.repository.EventRepository;
import com.neusoft.databus.policy.service.EventService;
import com.neusoft.databus.policy.utils.EventCountUtil;
import com.neusoft.databus.policy.utils.PolicyUtil;

@Service
@Transactional(rollbackFor = Exception.class)
public class EventServiceImpl implements EventService {
    @Autowired
    private EventRepository eventRepository;
    @Autowired
    private EventAttibuteRepository eventAttibuteRepository;
    @Autowired
    private EventConverter eventConverter;
    @Autowired
    private CacheService cacheService;
    @Autowired
    private ProducerService producerService;
    @Value("${databus.policy.eventTopic}")
    private String topic;
    private final String cacheName = DatabusPolicyConstant.POLICY_EVENT_CACHE_NAME;
    @Autowired
    private AsyncEventBus policyEventBus;
    @Autowired
    private EventCountUtil eventCountUtil;

    @PostConstruct
    public void init() {
        this.policyEventBus.register(this);
    }

    @Override
    public List<Event> findByStatus(final EventStatus eventStatus) {
        return this.eventRepository.findByStatus(eventStatus);
    }

    @Override
    public Event create(final Event model) {
        return this.eventRepository.save(model);
    }

    @Override
    public void delete(final Event model) {
        this.eventRepository.delete(model);
    }

    @Override
    public List<Event> findByResourceCodeAndStatus(final String resourceCode, final EventStatus status) {
        return this.eventRepository.findByResourceCodeAndStatus(resourceCode, status);
    }

    @Override
    public List<Event> findByPolicyIdAndStatus(final Long policyId, final EventStatus status) {
        return this.eventRepository.findByPolicyIdAndStatus(policyId, status);
    }

    @Override
    public void deleteById(final Long id) {
        this.eventRepository.deleteById(id);
    }

    @Override
    public Event update(final Event model) {
        return this.eventRepository.save(model);
    }

    @Override
    public Event eventTrigger(final Event model) {
        final List<Event> events = this.eventRepository.findByCodeAndStatus(model.getCode(), EventStatus.UNFIXED);
        if ((null == events) || (events.size() == 0)) {
            model.getEventAttribute().setCount(1L);
            this.sendMessage(this.eventRepository.save(model));
        } else {
            events.forEach(e -> {
                final EventAttribute eventAttibute = e.getEventAttribute();
                eventAttibute.setContent(model.getEventAttribute().getContent());
                eventAttibute.setLastTime(model.getEventAttribute().getLastTime());
                eventAttibute.setCount(e.getEventAttribute().getCount() + 1);
                this.eventAttibuteRepository.save(eventAttibute);
            });
        }
        return model;
    }

    @Override
    public void eventFixed(final Event model) {
        if (null != model) {
            final Date now = new Date();
            if (null != model.getFixTime()) {
                final List<Event> events =
                        this.eventRepository.findByCodeAndStatus(model.getCode(), EventStatus.UNFIXED);
                events.forEach(e -> {
                    e.setFixTime(model.getFixTime());
                    this.eventFix(e);
                });
            } else {
                model.setUpdateTime(now);
                model.setFixTime(now);
                this.eventFix(model);
            }
        }
    }

    @Override
    public void eventFixed(final List<Event> models) {
        final Date now = new Date();
        models.forEach(e -> {
            // 解除告警事件
            e.setFixTime(now);
            e.setUpdateTime(now);
            this.eventFix(e);
        });
    }

    private Event eventFix(Event event) {
        event.setStatus(EventStatus.FIXED);
        event = this.eventRepository.save(event);
        // 删除缓存
        this.removeEventCache(event);
        // 发送消息
        this.sendMessage(event);
        return event;
    }

    /**
     * 删除告警事件相关缓存
     *
     * @param event
     */
    private void removeEventCache(final Event event) {
        final String resourceCode = event.getResourceCode();
        final Policy policy = event.getPolicy();
        String key = PolicyUtil.getEventKey(resourceCode, policy.getCode());
        this.removeCache(this.cacheName, key);
        //禁用策略
        if (event.getPolicy().isDisabled()) {
            //历史数据缓存
            key = PolicyUtil.getOldMetricValueKey(resourceCode, policy.getId(),
                    event.getMetricCode());
            String cacheName = DatabusPolicyConstant.POLICY_EXPRESSION_CACHE_NAME;
            this.removeCache(cacheName, key);
            //频次缓存
            cacheName = DatabusPolicyConstant.POLICY_FREQUENCY_CACHE_NAME;
            key = PolicyUtil.getFrequencyKey(resourceCode, policy);
            this.removeCache(cacheName, key);
            //复杂告警
            if (policy.getPolicyType().equals(PolicyType.POLICY_COMPLEX)) {
                final String complexCacheName = DatabusPolicyConstant.POLICY_EXPRESSION_CACHE_NAME;
                final String complexPolicyKey = PolicyUtil.getComplexPolicyKey(resourceCode, policy);
                this.removeCache(complexCacheName, complexPolicyKey);
                List<ComplexPolicyExtendBean> complexPolicyExtendBeans = Lists.newArrayList();
                complexPolicyExtendBeans = JsonUtils.jsonToPojoList(policy.getExtendProperty(),
                        new TypeReference<List<ComplexPolicyExtendBean>>() {
                        });
                complexPolicyExtendBeans.forEach(e -> {
                    final String historyKey = PolicyUtil.getHistoryKey(e);
                    this.removeCache(complexCacheName, historyKey);
                });
            }
        }
    }

    /**
     * 清空缓存
     *
     * @param cacheName
     * @param key
     */
    private void removeCache(final String cacheName, final String key) {
        if (this.cacheService.isInCache(cacheName, key)) {
            this.cacheService.remove(cacheName, key);
        }
    }

    //    @AllowConcurrentEvents // 线程安全
    @Subscribe // 异步执行的方法标识:需要传入Event类型参数
    public void processPolicyEvent(final Event model) {
        if (EventStatus.FIXED.equals(model.getStatus())) {
            this.eventFixed(model);
        } else if (EventStatus.UNFIXED.equals(model.getStatus())) {
            this.eventTrigger(model);
        }
    }

    /**
     * 告警发送mq
     *
     * @param event
     */
    private void sendMessage(final Event event) {
        // 发送消息,默认使用kafka
        final MessageBean message = new MessageBean(JsonUtils.pojoToJson(this.eventConverter.toVO(event)), new Date());
        this.producerService.send(this.topic, message);
        //告警数量
        this.eventCountUtil.eventCount(event.getPolicy().getPolicyType(), event.getStatus());
    }

    @Override
    public Integer eventCount(final PolicyType policyType) {
        final Integer eventCount = this.eventCountUtil.eventCount(policyType);
        return eventCount;
    }

}
